package com.deep.flink;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;
import java.util.List;
import java.util.Map;

public class LoginFailDetectExample {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 1. 获取登录事件 user1登录失败
        SingleOutputStreamOperator<LogEvent> logEventStream = env.fromElements(
                new LogEvent("user_1", "192.168.0.1", "fail", 2000L),
                new LogEvent("user_1", "192.168.0.2", "fail", 3000L),
                new LogEvent("user_2", "192.168.1.29", "fail", 4000L),
                new LogEvent("user_1", "192.168.23.10", "fail", 5000L),
                new LogEvent("user_2", "192.168.1.29", "fail", 7000L),
                new LogEvent("user_2", "192.168.1.29", "fail", 8000L),
//                new LogEvent("user_2", "192.168.1.29", "success", 6000L)
//                new LogEvent("user_2", "192.168.1.29", "fail", 8000L)
//                new LogEvent("user_2", "192.168.1.29", "success", 8000L)
                new LogEvent("user_2", "192.168.1.29", "fail", 6000L)

        ).assignTimestampsAndWatermarks(WatermarkStrategy.<LogEvent>forBoundedOutOfOrderness(Duration.ZERO) // Duration.ZERO Duration.ofSeconds(3)
                .withTimestampAssigner(new SerializableTimestampAssigner<LogEvent>() {
                    @Override
                    public long extractTimestamp(LogEvent logEvent, long l) {
                        return logEvent.timestamp;
                    }
                }));

        // 2. 复杂事件模式 ， 练习登录失败3次
        Pattern<LogEvent, LogEvent> patWhere = Pattern.<LogEvent>begin("first")
                .where(new SimpleCondition<LogEvent>() {
                    @Override
                    public boolean filter(LogEvent logEvent) throws Exception {
                        return "fail".equals(logEvent.eventType);
                    }
                })
                .next("second")
                .where(new SimpleCondition<LogEvent>() {
                    @Override
                    public boolean filter(LogEvent logEvent) throws Exception {
                        return "fail".equals(logEvent.eventType);
                    }
                })
                .next("third")
                .where(new SimpleCondition<LogEvent>() {
                    @Override
                    public boolean filter(LogEvent logEvent) throws Exception {
                        return "fail".equals(logEvent.eventType);
                    }
                });

        // 3， 模式应用，检测复杂事件
        PatternStream<LogEvent> patternStream = CEP.pattern(logEventStream.keyBy(
                event -> event.userId
        ), patWhere);

        // 4. 检测结果，提取出来 ,进行处理，得到报警信息
        SingleOutputStreamOperator<String> warningStream = patternStream.select(new PatternSelectFunction<LogEvent, String>() {
            @Override
            public String select(Map<String, List<LogEvent>> map) throws Exception {

                System.out.println(map);
                System.out.println(map.keySet());
                // 提取 复杂事件中的三次登录失败事件
                LogEvent first = map.get("first").get(0);
                LogEvent second = map.get("second").get(0);
                LogEvent third = map.get("third").get(0);

                // 处理 返回
                return first.userId + "连续3次登录失败！登录时间：" +
                        "" + first.timestamp + "" +
                        "" + second.timestamp + "" +
                        "" + third.timestamp + "";
            }
        });
        //
        warningStream.print();

        env.execute("select");

    }
}
